package com.atguigu.flink.edu.app.dws;

import com.alibaba.druid.util.StringUtils;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.flink.edu.beans.TrafficPageViewBean;
import com.atguigu.flink.edu.utils.DataFormatUtil;
import com.atguigu.flink.edu.utils.MyClickhouseUtil;
import com.atguigu.flink.edu.utils.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;

public class DwsTrafficVcChIsNewPageViewWindow {
    public static void main(String[] args) throws Exception {
        //TODO 基本环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        //TODO 检查点相关设置
        //TODO 从kafka的页面日志主题中读取数据
        String topic = "dwd_traffic_page_log";
        String groupId = "dws_traffic_vc_ch_ar_isnew_group";
        KafkaSource<String> kafkaSource = MyKafkaUtil.getKafkaSource(topic, groupId);
        DataStreamSource<String> kafkaDS = env
                .fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafka_source");

        //TODO 对数据进行类型转换
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaDS
                .map(new MapFunction<String, JSONObject>() {
                    @Override
                    public JSONObject map(String s) throws Exception {
                        JSONObject jsonObject = JSON.parseObject(s);
                        return jsonObject;
                    }
                });

        //TODO 对流中数据按设备id分组
        KeyedStream<JSONObject, String> keyedDS = jsonObjDS
                .keyBy(jsonObject -> jsonObject.getJSONObject("common").getString("mid"));

        //TODO 对度量值进行计数  并将结果封装为实体类对象
        SingleOutputStreamOperator<TrafficPageViewBean> beanDS = keyedDS
                .process(new KeyedProcessFunction<String, JSONObject, TrafficPageViewBean>() {
                    private ValueState<String> lastVisitDateState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        ValueStateDescriptor<String> valueStateDescriptor = new ValueStateDescriptor<>(
                                "valuestate", String.class);
                        // 独立访客每日重置，所以状态最多维护一天即可，为了减少内存开销，设置状态ttl为1天。
                        valueStateDescriptor.enableTimeToLive(StateTtlConfig.newBuilder(Time.days(1)).build());
                        lastVisitDateState = getRuntimeContext().getState(valueStateDescriptor);
                    }

                    @Override
                    public void processElement(JSONObject jsonObj, Context ctx, Collector<TrafficPageViewBean> out) throws Exception {
                        // 获取日志中的维度属性，赋值给实体类的同名字段。
                        JSONObject commonObj = jsonObj.getJSONObject("common");
                        JSONObject pageObj = jsonObj.getJSONObject("page");
                        String vc = commonObj.getString("vc");
                        String ch = commonObj.getString("ch");
                        String ar = commonObj.getString("ar");
                        String isNew = commonObj.getString("is_new");
                        String pageId = pageObj.getString("page_id");
                        Long duringTime = pageObj.getLong("during_time");
                        Long ts = jsonObj.getLong("ts");
                        System.out.println("==>"+ts);
                        // 如果状态中的日期为null或不为当日，则将独立访客数置为1，否则置为0。
                        String lastVisitDate = lastVisitDateState.value();
                        String curDate = DataFormatUtil.toDate(ts);
                        Long uvCt = 0L;
                        if (StringUtils.isEmpty(lastVisitDate) || !curDate.equals(lastVisitDate)) {
                            uvCt = 1L;
                            //最后在状态中维护末次访问日期，
                            lastVisitDateState.update(curDate);
                        }
                        // 判断 last_page_id 是否为 null，如果是，说明页面是首页，开启了一个新的会话，将会话数置为 1，否则置为 0。
                        Long svCt = "home".equals(pageId)? 1L : 0L;
                        //只要有一条页面浏览日志，页面浏览数加一
                        Long pvCt = 1L;
                        //补充维度字段，窗口起始和结束时间置为空字符串。
                        //下游要根据水位线开窗，所以要补充事件时间字段，此处将日志生成时间 ts 作为事件时间字段即可。
                        TrafficPageViewBean pageViewBean = new TrafficPageViewBean(
                                "", "", vc, ch, ar, isNew, uvCt, svCt, pvCt, duringTime, ts);
                        out.collect(pageViewBean);
                        System.out.println("===>>"+pageViewBean.getTs());
                    }
                });

        //TODO 制定watermark及提取事件时间字段
        SingleOutputStreamOperator<TrafficPageViewBean> watermarkDS = beanDS
                .assignTimestampsAndWatermarks(WatermarkStrategy
                        .<TrafficPageViewBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        .withTimestampAssigner(new SerializableTimestampAssigner<TrafficPageViewBean>() {
                            @Override
                            public long extractTimestamp(TrafficPageViewBean trafficPageViewBean, long l) {
                                return trafficPageViewBean.getTs();
                            }
                        }));
        //TODO 分组 ：按照维度字段分组
        KeyedStream<TrafficPageViewBean, Tuple4<String, String, String, String>> keyedDimDs = watermarkDS
                .keyBy(new KeySelector<TrafficPageViewBean, Tuple4<String, String, String, String>>() {
                    @Override
                    public Tuple4<String, String, String, String> getKey(TrafficPageViewBean pvBean) throws Exception {
                        return Tuple4.of(pvBean.getVc(), pvBean.getCh(), pvBean.getAr(), pvBean.getIsNew());
                    }
                });

        //TODO 开窗
        WindowedStream<TrafficPageViewBean, Tuple4<String, String, String, String>, TimeWindow> windowDS = keyedDimDs
                .window(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10)));

        //TODO 聚合计算
        SingleOutputStreamOperator<TrafficPageViewBean> reduceDS = windowDS.reduce(
                new ReduceFunction<TrafficPageViewBean>() {
                    @Override
                    public TrafficPageViewBean reduce(TrafficPageViewBean pvBean1, TrafficPageViewBean pvBean2) throws Exception {
                        pvBean1.setUvCt(pvBean1.getUvCt() + pvBean2.getUvCt());
                        pvBean1.setPvCt(pvBean1.getPvCt() + pvBean2.getPvCt());
                        pvBean1.setSvCt(pvBean1.getSvCt() + pvBean2.getSvCt());
                        pvBean1.setDurSum(pvBean1.getDurSum() + pvBean2.getDurSum());
                        return pvBean1;
                    }
                },
                new WindowFunction<TrafficPageViewBean, TrafficPageViewBean, Tuple4<String, String, String, String>, TimeWindow>() {
                    @Override
                    public void apply(Tuple4<String, String, String, String> stringTuple4, TimeWindow window, Iterable<TrafficPageViewBean> input, Collector<TrafficPageViewBean> out) throws Exception {
                        for (TrafficPageViewBean viewBean : input) {
                            viewBean.setStt(DataFormatUtil.toYmdHms(window.getStart()));
                            System.out.println(window.getStart());
                            viewBean.setEdt(DataFormatUtil.toYmdHms(window.getEnd()));
                            viewBean.setTs(System.currentTimeMillis());
                            out.collect(viewBean);
                        }
                    }
                });

        //TODO 将聚合结果写到ck中
        reduceDS.print(">>>");
        reduceDS.addSink(MyClickhouseUtil
                .getSinkFunction("insert into dws_traffic_vc_ch_ar_is_new_page_view_window values(?,?,?,?,?,?,?,?,?,?,?)"));

        env.execute();
    }
}
